package flinkdemo.kafak;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.db.Db;
import cn.hutool.db.Entity;
import cn.hutool.db.ds.DSFactory;
import cn.hutool.setting.Setting;
import com.arcsoft.face.FaceFeature;
import com.arcsoft.face.FaceInfo;
import com.arcsoft.face.FaceSimilar;
import com.arcsoft.face.enums.DetectMode;
import com.arcsoft.face.enums.DetectOrient;
import com.arcsoft.face.enums.ImageFormat;
import com.google.common.collect.Lists;
import flinkdemo.kafak.conf.EngineConfigurationSer;
import flinkdemo.kafak.entity.FindFaceInfo;
import flinkdemo.kafak.entity.ProcessInfo;
import flinkdemo.kafak.entity.VideoMsg;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;

public class FaceProcessWindowFunction extends ProcessWindowFunction<Tuple2<String, VideoMsg>, List<VideoMsg>, Tuple, GlobalWindow> {

	public int passRate;
	private String appId = "BGFUM4yT5M9PRGfd9VgUXMzGbK1jZJer4YJzG77HovXR";
	private String sdkKey = "AGURJP2c7PRahrMsS9fPFqW6dFGrgXpdQpELRFEyhtoS";

	private Integer detectFaceMaxNum=10;
	private Integer detectFaceScaleVal=16;
	private DetectMode detectMode= DetectMode.ASF_DETECT_MODE_IMAGE;
	private DetectOrient detectFaceOrientPriority= DetectOrient.ASF_OP_0_ONLY;
	private FunctionConfigurationSer functionConfiguration;

	private EngineConfigurationSer engineConfiguration;

	private FaceEngine faceEngine = new FaceEngine();

	private List<Pair<String,Object>> hashMap;

	FaceProcessWindowFunction(String appId, String sdkKey,int passRate) throws Exception {
		this.appId = appId;
		this.sdkKey = sdkKey;
		this.passRate = passRate;
		this.hashMap = facesInDb();
	}

	FaceProcessWindowFunction() throws Exception {
		this.hashMap = facesInDb();
	}

	public List<Pair<String,Object>> facesInDb() throws Exception {

		List<Pair<String,Object>> hashMap = new ArrayList<>();
		//自定义数据库Setting，更多实用请参阅Hutool-Setting章节
		Setting setting = new Setting("flinkface/db.setting");
		//获取指定配置，第二个参数为分组，用于多数据源，无分组情况下传null
		DataSource ds = DSFactory.create(setting).getDataSource();

		List<Entity> faceFeatures = Db.use(ds).query("select FACE_ID,FACE_FEATURE from user_face_info where checked = '0' ");
		if (CollectionUtils.isNotEmpty(faceFeatures)){
			for (Entity face : faceFeatures){
				hashMap.add(Pair.of(String.valueOf(face.get("FACE_ID")), face.get("FACE_FEATURE")));
			}
		}
		return hashMap;
	}

	@Override
	public void process(Tuple tuple, Context context, Iterable<Tuple2<String,VideoMsg>> iterable, Collector<List<VideoMsg>> collector) throws Exception {
		List<VideoMsg> vs = new ArrayList<>();
		for (Tuple2<String,VideoMsg> in :iterable){

			VideoMsg imageInfo = in.f1;
			List<ProcessInfo> infos = extractFacesFeature(imageInfo);
			//筛选出有人脸的
			if (CollectionUtils.isEmpty(infos) || CollectionUtils.isEmpty(hashMap)) {
				return;
			}
			// 一张人脸图片上有 多个人脸，每个人脸 有多个相似脸
			List<List<FindFaceInfo>> finds = new ArrayList<>();
			for(ProcessInfo info : infos){
				List<FindFaceInfo> find = new ArrayList<>();
				for(Pair<String,Object> fc : hashMap ){
					FaceSimilar similar = new FaceSimilar();
					FaceFeature dBFeatrue = new FaceFeature((byte[]) fc.getRight());
					faceEngine.compareFaceFeature(info.getFaceFeature(),dBFeatrue,similar);
					// 获取相似值
					int similarValue = plusHundred(similar.getScore());
					// 相似值大于配置预期，加入到识别到人脸的列表
					if (similarValue > passRate) {
						FindFaceInfo finfo = new FindFaceInfo();
						finfo.setFaceId(info.getFace().getFaceId());
						finfo.setOrient(info.getFace().getOrient());
						finfo.setRect(info.getFace().getRect());
						finfo.setSimilarValue(similarValue);
						find.add(finfo);
						//log.debug("task线程：" + Thread.currentThread().getName() + "相似度=" + similarValue);
					}
				}
				finds.add(find);
				find.clear();
			}
			//imageInfo.setInfos(finds);
			//imageInfo.setFind(CollectionUtils.isEmpty(finds)?0:1);
			//清理到人脸识别的数据
			imageInfo.setRgb(null);
			vs.add(imageInfo);
		}
		collector.collect(vs);
	}

	private List<ProcessInfo> extractFacesFeature(VideoMsg imageInfo) {
		if(imageInfo == null) {
			return null;
		}
		try{
			// 人脸检测得到人脸列表
			List<FaceInfo> facesInImg = Lists.newArrayList();
			// 图片的长宽和数据
			final byte[] imgByts = imageInfo.getRgb();
			final int width = imageInfo.getWidth();
			final int height = imageInfo.getHeight();
			//int code =
			faceEngine.detectFaces(imgByts, width, height, ImageFormat.CP_PAF_BGR24, facesInImg);

			if (CollectionUtil.isEmpty(facesInImg)) {
				return null;
			}

			List<ProcessInfo> features = Lists.newArrayList();
			for (FaceInfo face : facesInImg) {
				FaceFeature faceFeature = new FaceFeature();
				faceEngine.extractFaceFeature(imgByts, width, height, ImageFormat.CP_PAF_BGR24, face, faceFeature);
				if (faceFeature.getFeatureData() != null){
					ProcessInfo info = new ProcessInfo();
					info.setFace(face);
					info.setFaceFeature(faceFeature);
					features.add(info);
				}
			}
			return features;
		} catch (Exception e) {
			e.printStackTrace();

		}
		return null;
	}

	/** 乘以100 */
	private int plusHundred(Float value) {
		return new BigDecimal(value).multiply(new BigDecimal(100f)).intValue();
	}




	@Override
	public void open(Configuration parameters) throws Exception {
		faceEngine.activeOnline(appId,sdkKey);
		engineConfiguration = new EngineConfigurationSer();

		functionConfiguration = new FunctionConfigurationSer();
		functionConfiguration.setSupportFaceDetect(true);
		functionConfiguration.setSupportFaceRecognition(true);

		engineConfiguration.setFunctionConfiguration(functionConfiguration);
		engineConfiguration.setDetectFaceMaxNum(detectFaceMaxNum);
		engineConfiguration.setDetectFaceScaleVal(detectFaceScaleVal);
		engineConfiguration.setDetectMode(detectMode);
		engineConfiguration.setDetectFaceOrientPriority(detectFaceOrientPriority);

		faceEngine.init(engineConfiguration);
		super.open(parameters);
	}

	@Override
	public void close() throws Exception {
		super.close();
	}
}